package pack01_udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

/*
 "zhangsan:18,lisi:20,wangwu:17"
 */
public class MyUDTF2 extends GenericUDTF {
    private final  Object[] forwardListObj = new Object[2];

    /*
       1、该方法只会执行一次，用来做初始化
       2、该方法用来指定你转换完之后每一列名字和每一列的类型
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        //设置列的名字
        List<String> fieldNames = new ArrayList();

        //设置列名，就是转换完之后列名字
        fieldNames.add("name");
        fieldNames.add("age");

        //设置列的类型
        List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>()  ;
        //设置输出的列的值类型
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector); //String类型
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector); //int 类型

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

    }


    /*
       1、表中的数据有多少行需要处理，该方法就会执行多少次
       2、函数最后如何调用
         select my_explode2('zhangsan:18,lisi:20,wangwu:17',',',':');
            objects[0] = 'zhangsan:18,lisi:20,wangwu:17'
            objects[1] = ','
            objects[2] = ':'
     */
    @Override
    public void process(Object[] objects) throws HiveException {
        //1:获取原始数据
        String  lineStr =     objects[0].toString(); // "zookeeper,hadoop,hdfs,hive,MapReduce"

        //2:获取数据传入的第二个参数，此处为分隔符
        String firstSeparator = objects[1].toString();  // ","

        //3:获取数据传入的第三个参数，此处为分隔符
        String secondeSeparator = objects[2].toString();  // ":"


        //3.将原始数据按照传入的第一个分隔符(,)进行切分
        String[] array1 = lineStr.split(firstSeparator); //["zhangsan:18","lisi:20","wangwu:17"]

        //4、遍历数组
        for (String keyValue : array1) {
            //5、将array1中的keyVlue数据按照传入的第二个分隔符(：)进行切分
            String[] array2 = keyValue.split(secondeSeparator); //["zhangsan",18]

            //6、将结果写出
            forwardListObj[0]=array2[0];
            forwardListObj[1]=Integer.parseInt(array2[1]);
            forward(forwardListObj);  //每执行一次forward，结果表就会多出一行
        }



    }

    @Override
    public void close() throws HiveException {

    }
}
